package com.min.gps.service.nio;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * NIO socket 客户端
 * @author cgli
 *
 */
public class SocketClient {
	private static final Logger logger = LoggerFactory.getLogger(SocketServer.class);
	public SocketClient() {
		try {
			client = SocketChannel.open();
			client.configureBlocking(false);
			String ip = "127.0.0.1";
			Integer port =1122;
			client.connect(new InetSocketAddress(ip, port));
			client.register(selector, SelectionKey.OP_CONNECT);
		} catch (IOException e) {
			logger.info("创建客户端连接异常Client2" + e.getMessage());
			close();
		}
	}

	private SocketChannel client;
	private Selector selector = getSelector();
	private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
			new ArrayBlockingQueue<Runnable>(20));

	private volatile boolean isClose = false;

	private AtomicLong writeCount = new AtomicLong(0);
	private AtomicLong readCount = new AtomicLong(0);

	private AtomicBoolean isWriting = new AtomicBoolean(false);
	private AtomicBoolean isReading = new AtomicBoolean(false);

	public Selector getSelector() {
		try {
			return Selector.open();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	public void start() {
		threadPool.execute(new SelectorGuardHandler());
	}

	private class SelectorGuardHandler implements Runnable {

		@Override
		public void run() {

			while (!isClose) {
				try {
					if (selector.select(10) == 0) {
						continue;
					}
					Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
					while (iterator.hasNext()) {
						SelectionKey selectionKey = iterator.next();
						iterator.remove();
						if (selectionKey.isReadable()) {
							SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
							if (isReading.get()) {
								Thread.sleep(5);
							} else {
								isReading.set(true);
								threadPool.execute(new ReceiveMessageHandler(socketChannel));
							}

						} else
							if (selectionKey.isWritable()) {
							Object requestMessage = selectionKey.attachment();
							SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
							selectionKey.interestOps(SelectionKey.OP_READ);
							threadPool.execute(new SendMessageHandler(socketChannel, requestMessage));
						} else if (selectionKey.isConnectable()) {
							SocketChannel sc = (SocketChannel) selectionKey.channel();
							sc.finishConnect();
							sc.register(selector, SelectionKey.OP_READ);
						}
					}
				} catch (Exception e) {
					logger.info("客户端启动或运行异常[start]：" + e.getMessage());
					close();
				}
			}
		}
	}

	/**
	 * 发送数据线程
	 * 
	 */
	private class SendMessageHandler implements Runnable {
		private SocketChannel client;

		private Object requestMessage;

		private SendMessageHandler(SocketChannel client, Object requestMessage) {
			this.client = client;
			this.requestMessage = requestMessage;
		}

		@Override
		public void run() {
			try {
				byte[] requestMessageByteData = null;
				if (requestMessage instanceof byte[]) {
					requestMessageByteData = (byte[]) requestMessage;
				} else if (requestMessage instanceof String) {
					requestMessageByteData = ((String) requestMessage).getBytes();
				}
				if (requestMessageByteData == null || requestMessageByteData.length == 0) {
					logger.info("客户端发送的数据为空");
				} else {

					ByteBuffer data = ByteBuffer.allocate(requestMessageByteData.length + 4);
					data.putInt(requestMessageByteData.length);
					data.put(requestMessageByteData);
					data.flip();
					while (data.hasRemaining()) {
						client.write(data);
					}
					Date date = new Date();
					SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS");
					logger.info("[" + sdf.format(date) + "][" + Thread.currentThread().getId() + "]客户端发送数据：["
							+ new String(requestMessageByteData) + "]");
				}
			} catch (Exception e) {
				logger.info("客户端发送数据失败：[" + e.getMessage() + "]");
				close();
			} finally {
				isWriting.set(false);
				writeCount.decrementAndGet();
			}

		}
	}

	/**
	 * 读数据线程
	 */
	private class ReceiveMessageHandler implements Runnable {
		private SocketChannel client;
		private ByteBuffer dataLen = ByteBuffer.allocate(4);

		private ReceiveMessageHandler(SocketChannel client) {
			this.client = client;
		}

		@Override
		public void run() {
			try {
				dataLen.clear();
				int len = 4;
				while (len > 0) {
					int readLen = client.read(dataLen);
					if (readLen < 0) {
						throw new Exception("readLen==" + readLen);
					} else if (readLen == 0) {
						logger.info(Thread.currentThread().getId() + "readLen == 0");
						return;
					}
					len -= readLen;
				}
				int data_length = dataLen.getInt(0);
				ByteBuffer data = ByteBuffer.allocate(data_length);
				while (data.hasRemaining()) {
					client.read(data);
				}

				String receiveData = new String(data.array());
				Date date = new Date();
				SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS");
				logger.info("[" + sdf.format(date) + "][" + Thread.currentThread().getId() + "]客户端接收到服务器["
						+ client.getRemoteAddress() + "]数据 :[" + receiveData + "]");
				readCount.incrementAndGet();
			} catch (Exception e) {
				logger.info("客户端接收数据失败：" + e);
				close();
			} finally {
				isReading.set(false);
			}
		}
	}

	public boolean isClose() {
		return isClose;
	}

	public void setClose(boolean close) {
		this.isClose = close;
	}

	public void close() {
		try {
			threadPool.shutdown();
			isClose = true;
			if (selector != null) {
				selector.close();
			}
			if (client != null) {
				client.close();
			}
		} catch (IOException e) {
			logger.info("关闭客户端异常：" + e.getMessage());
		}
	}

//	public void writeData(String data) {
//		while (isWriting.get()) {
//			try {
//				Thread.sleep(5);
//			} catch (InterruptedException e) {
//				e.printStackTrace();
//			}
//		}
//		try {
//			isWriting.set(true);
//			writeCount.incrementAndGet();
//			client.register(selector, SelectionKey.OP_WRITE, data);
//		} catch (Exception e) {
//			logger.info("客户端注册写通道异常：" + e.getMessage());
//		}
//	}
	public void writeData(byte[] data) {
		while (isWriting.get()) {
			try {
				Thread.sleep(5);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		try {
			isWriting.set(true);
			writeCount.incrementAndGet();
			client.register(selector, SelectionKey.OP_WRITE, data);
		} catch (Exception e) {
			logger.info("客户端注册写通道异常：" + e.getMessage());
		}
	}

	public boolean hasWriteTask() {
		return writeCount.get() != 0;
	}

	public long getRecive() {
		return readCount.get();
	}

	public static void main(String[] args) {

		for (int j = 0; j < 10; j++) {
			new Thread(new Runnable() {

				@Override
				public void run() {

					final SocketClient client = new SocketClient();
					client.start();
					try {
						TimeUnit.MILLISECONDS.sleep(500);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					long t1 = System.currentTimeMillis();
//					for (int i = 0; i < 500; i++) {
//						String ii = "00" + i;
//						ii = ii.substring(ii.length() - 3);
//						client.writeData(ii + "ccccccccccccccccccc发送数据ccccccccdddddddddddd" + i);
//					}
					for (int i = 0; i < 100; i++) {
						byte[] loginBytes =new byte[] {(byte)0xAF,0x00,0x56,(byte)0xAF,0x00,0x01,0x01,0x24,0x47,0x4E,0x52,0x4D,0x43,0x2C,0x30,0x36,0x32,0x33,0x31,0x38,0x2E,0x38,0x32,0x33,0x2C,0x41,0x2C,0x32,0x32,0x37,0x37,0x2E,0x31,0x30,0x38,0x31,0x33,0x2C,0x4E,0x2C,0x31,0x31,0x33,0x31,0x39,0x2E,0x39,0x34,0x37,0x35,0x34,0x30,0x30,0x30,0x30,0x30,0x30,0x31,0x2C,0x45,0x2C,0x32,0x31,0x2E,0x33,0x37,0x34,0x2C,0x39,0x31,0x2E,0x37,0x37,0x2C,0x32,0x39,0x30,0x36,0x31,0x37,0x2C,0x2C,0x2C,0x44,0x2A,0x36,0x37,0x0D,0x0A,0x2D,0x3F,(byte)0xD3,(byte)0xFF};
						//byte[] loginBytes =new byte[] {(byte)  0xAF,0x00,0x75,(byte) 0xAF,0x00,0x0A,0x02,0x55,0x00,0x02,0x09,0x00,0x60,0x55,0x00,0x02,0x03,0x00,0x5A,0x55,0x00,0x02,0x0C,0x00,0x63,0x55,0x00,0x02,0x04,0x00,0x5B,0x55,0x00,0x02,0x01,0x00,0x58,0x55,0x00,0x02,0x16,0x00,0x6D,0x55,0x00,0x02,0x12,0x00,0x69,0x55,0x00,0x02,0x07,0x00,0x5E,0x55,0x00,0x02,0x0A,0x00,0x61,0x55,0x00,0x02,0x02,0x00,0x59,0x55,0x00,0x02,0x08,0x00,0x5F,0x55,0x00,0x02,0x05,0x00,0x5C,0x55,0x00,0x02,0x10,0x00,0x67,0x55,0x00,0x02,0x13,0x00,0x6A,0x55,0x00,0x02,0x14,0x00,0x6B,0x55,0x00,0x02,0x0E,0x00,0x65,0x55,0x00,0x02,0x0D,0x00,0x64,0x55,0x00,0x02,0x0B,0x00,0x62,0x55,0x00,0x02,0x0F,0x00,0x66,(byte) 0xC4,0x65,(byte) 0xFF};
						client.writeData(loginBytes);
					}
					while (client.hasWriteTask()) {
						try {
							Thread.sleep(10);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
					while (client.getRecive() != 500) {
						try {
							Thread.sleep(10);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
					long t2 = System.currentTimeMillis();
					logger.info("总共耗时：" + (t2 - t1) + "ms");
					client.close();
				}

			}).start();
		}
	}

}
